package com.mj.order;

import com.mj.utils.Util;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 顺序消息，同一个queue是顺序消费，但是不同的queue是不顺序的，取决于producer发送时候选择的queue
 */
public class OrderConsumer {
    public static void main(String[] args) throws MQClientException {
        // 1.和生产者一样，创建一个DefaultMQPushConsumer实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Util.ConsumerGroup);
        // 2.设置NameServer的地址，消费者通过这个地址与NameServer进行通信，来获取Broker的地址信息
        consumer.setNamesrvAddr(Util.NamesrvAddr);
        // 订阅一个或多个Topic，以及Tag来过滤需要消费的消息。我们订阅了"MyTopic1"，使用"*"来匹配此Topic下的所有Tag
        consumer.subscribe(Util.Topic, "*");
        // 3.注册消息监听器，用于处理从Broker接收到的消息。使用MessageListenerOrderly接口的实现，按queue顺序消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            // 4.当收到消息时，方法会被调用。
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                // 5.遍历消息列表，并打印每条消息的内容（注意：这里直接打印msg对象不会得到预期的消息内容字符串）
                for (MessageExt msg : list) {
                    // 所以我们打印msg.getBody()的内容，为了保留消息原样
                    System.out.println("已收到消息" + msg);
                }
                // 6.返回消费状态，这里表示消息已成功消费
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 7.启动消费者实例
        consumer.start();
        // 8.打印日志消费者已启动
        System.out.println("消费者已启动");

    }
}
